package com.slipper.websocket;

import com.slipper.core.gson.GsonUtil;
import com.slipper.core.message.Message;
import com.slipper.core.message.MessageUtil;
import com.slipper.core.message.Subscriber;
import com.slipper.util.Tool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.lang.NonNull;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;

public class WebsocketHandler extends TextWebSocketHandler {
    private final Logger logger = LoggerFactory.getLogger(WebsocketHandler.class);
    private final ConcurrentHashMap<String, Websocket> websocketMap = new ConcurrentHashMap<>();
    private boolean isMonitor = false;
    private Timer timer;

    @Override
    public void handleTextMessage(@NonNull WebSocketSession session, TextMessage textMessage) {
        if (Tool.isBlank(textMessage.getPayload())) return;

        WebsocketMessage message = GsonUtil.fromJson(textMessage.getPayload(), WebsocketMessage.class);
        if (Tool.isBlank(message.getType())) return;

        Websocket websocket = websocketMap.get(session.getId());
        if (websocket == null) return;

        if (message.getType().trim().equals("heartbeat")) {
            try {
                session.sendMessage(new TextMessage(GsonUtil.toJson(new Message("heartbeat", "pong"))));
            } catch (Exception e) {
                logger.error("响应心跳失败！", e);
            }
        } else if (message.getType().trim().startsWith("subscribe.")) {
            String topic = message.getType().substring(message.getType().indexOf(".") + 1).trim();
            if (Tool.isBlank(topic)) return;
            Subscriber subscriber = websocket.computeIfAbsent();
            subscriber.addTopic(topic);
            MessageUtil.subscribe(topic, subscriber);
            logger.info("用户【{}】订阅主题：{}", session.getAttributes().get("userName"), topic);
        } else if (message.getType().trim().startsWith("unsubscribe.")) {
            String topic = message.getType().substring(message.getType().indexOf(".") + 1).trim();
            if (Tool.isBlank(topic)) return;
            Subscriber subscriber = websocket.computeIfAbsent();
            subscriber.removeTopic(topic);
            if (!subscriber.hasTopic()) {
                MessageUtil.unsubscribe(subscriber);
            }
            logger.info("用户【{}】取消订阅主题：{}", session.getAttributes().get("userName"), topic);
        }
    }

    @Override
    public void afterConnectionEstablished(@NonNull WebSocketSession session) {
        websocketMap.put(session.getId(), new Websocket(session));
        startMonitor();
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, @NonNull CloseStatus status) {
        Websocket websocket = websocketMap.remove(session.getId());
        if (websocket == null) return;
        MessageUtil.unsubscribe(websocket.computeIfAbsent());
    }

    private synchronized void startMonitor() {
        if (this.isMonitor) return;

        this.isMonitor = true;
        this.timer = new Timer();
        this.timer.schedule(new TimerTask() {
            public void run() {
                int clearCount = 0;
                for (String key : websocketMap.keySet()) {
                    Websocket websocket = websocketMap.get(key);
                    if (websocket != null && !websocket.getSession().isOpen()) {
                        clearCount++;
                        websocketMap.remove(key);
                        MessageUtil.unsubscribe(websocket.computeIfAbsent());
                    }
                }

                if (clearCount > 0) {
                    logger.info("当前WebSocket连接数：{}, 清理无效连接数：{}", websocketMap.size(), clearCount);
                }

                if (websocketMap.isEmpty()) {
                    stopMonitor();
                }
            }
        }, 0, 60 * 1000L);
    }

    private synchronized void stopMonitor() {
        this.timer.cancel();
        this.timer = null;
        this.isMonitor = false;
    }
}
